package com.bigdata.spark.core.req

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * @author : ranzlupup
 * @date : 2023/3/2 14:17
 */
object MyHotCategoryTop10Analysis {
    def main(args: Array[String]): Unit = {
        val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("example1")
        val sc = new SparkContext(sparkConf)

        val rdd: RDD[String] = sc.textFile("datas/user_visit_action.txt")

        val categoryRDD: RDD[(String, Int)] = rdd.filter(
            line => {
                val datas: Array[String] = line.split("_")
                datas(6) != "-1"
            }
        )
            .map(
                line => {
                    val datas: Array[String] = line.split("_")
                    (datas(6), 1)
                }
            )
            .reduceByKey(_ + _)

        val orderActionRDD: RDD[String] = rdd.filter(
            line => {
                val datas: Array[String] = line.split("_")
                datas(8) != "null"
            }
        )

        val orderCountRDD: RDD[(String, Int)] = orderActionRDD.flatMap(
            action => {
                val datas: Array[String] = action.split("_")
                val cid: String = datas(8)
                val cids: Array[String] = cid.split(",")
                cids.map(id => (id, 1))
            }
        ).reduceByKey(_ + _)


        val payActionRDD: RDD[String] = rdd.filter(
            line => {
                val datas: Array[String] = line.split("_")
                datas(10) != "null"
            }
        )
        val payCountRDD: RDD[(String, Int)] = payActionRDD.flatMap(
            action => {
                val line: Array[String] = action.split("_")
                val cid: String = line(10)
                val cids: Array[String] = cid.split(",")
                cids.map(id => (id, 1))
            }
        ).reduceByKey(_ + _)


        val cogroupRDD: RDD[(String, (Iterable[Int], Iterable[Int], Iterable[Int]))] = categoryRDD.cogroup(orderCountRDD, payCountRDD)
        val resultRDD: RDD[(String, (Int, Int, Int))] = cogroupRDD.mapValues {
            case (category, order, pay) => {
                var categoryN = 0
                var orderN = 0
                var payN = 0
                val catIter: Iterator[Int] = category.iterator
                while (catIter.hasNext) {
                    categoryN = catIter.next()
                }

                val orderIter: Iterator[Int] = order.iterator
                while (orderIter.hasNext) {
                    orderN = orderIter.next()
                }
                val payIter: Iterator[Int] = pay.iterator
                while (payIter.hasNext) {
                    payN = payIter.next()
                }
                (categoryN, orderN, payN)
            }
        }
        resultRDD.sortBy(_._2, false).take(10).foreach(println)

        sc.stop()
    }
}
